import random
import time
import time
import hashlib
import os
from io import BytesIO

import pandas as pd
from django.core.cache import cache
from sqlalchemy import create_engine
import pymysql
from contextlib import contextmanager
from typing import Optional, Union, List, Tuple

from django.conf import settings
import datetime


# 生成唯一ID方案
def generate_mid():
    return str(int(time.time() * 1000)) + ''.join(random.choices('0123456789', k=6))

def df2sql(df, tablename, renamelistordict=None):
    """
    将df入库

    ----------
    Paramaters
    ----------

    df: pd.DataFrame
        待入库df数据
    tablename: str
        目标表名称
    renamelistordict: union{None, list, dict}
        df对应数据库目标表列命重命名数据
        默认以df列名为数据库对应列，也可传入顺序列名或字典
    """
    if renamelistordict:
        if isinstance(renamelistordict, list):
            renamelistordict = dict(zip(df.columns, renamelistordict))
        df.rename(columns=renamelistordict, inplace=True)
    username = settings.DATABASES["default"]["USER"]
    password = settings.DATABASES["default"]["PASSWORD"]
    host = settings.DATABASES["default"]["HOST"]
    port = settings.DATABASES["default"]["PORT"]
    database = settings.DATABASES["default"]["NAME"]
    engine = create_engine(f'mysql+pymysql://{username}:{password}@{host}:{port}/{database}')
    df.to_sql(tablename, con=engine, if_exists='append', index=False)

def executsql(sql, params=None, fetch=True):
    db = SQLExecutor()
    try:
        # 查询示例
        data = db.execute(sql, params, fetch)
        return data
    except DBConnectionError as e:
        print(f"Database error: {str(e)}")
    finally:
        db.close()


class DBConnectionError(Exception):
    """自定义数据库连接异常"""
    pass


class SQLExecutor:
    def __init__(self):
        self._conn: Optional[pymysql.Connection] = None
        self._cursor: Optional[pymysql.cursors.Cursor] = None

    @contextmanager
    def _connection(self):
        """上下文管理连接"""
        try:
            if self._conn is None or not self._conn.open:
                self._connect()

            yield self._cursor
            self._conn.commit()

        except pymysql.Error as e:
            self._conn.rollback()
            raise DBConnectionError(f"Database operation failed: {str(e)}") from e
        finally:
            self._close_cursor()

    def _connect(self):
        """建立新连接"""
        try:
            self._conn = pymysql.connect(
                host=DATABASES["default"]["HOST"],
                user=DATABASES["default"]["USER"],
                password=DATABASES["default"]["PASSWORD"],
                database=DATABASES["default"]["NAME"],
                autocommit=False,  # 显式事务控制
                charset='utf8mb4',
                cursorclass=pymysql.cursors.DictCursor  # 返回字典格式结果
            )
            self._cursor = self._conn.cursor()
        except pymysql.Error as e:
            raise DBConnectionError(f"Connection failed: {str(e)}") from e

    def _close_cursor(self):
        """关闭游标（保留连接）"""
        if self._cursor:
            self._cursor.close()
            self._cursor = None

    def close(self):
        """主动关闭连接"""
        self._close_cursor()
        if self._conn and self._conn.open:
            self._conn.close()
            self._conn = None

    def execute(
            self,
            sql: str,
            params: Union[List, Tuple, None] = None,
            fetch: bool = True
    ) -> Optional[pd.DataFrame]:
        """
        执行SQL语句

        :param sql: SQL语句
        :param params: 参数列表
        :param fetch: 是否获取结果
        :return: DataFrame或影响行数
        """
        with self._connection() as cursor:
            try:
                affected_rows = cursor.execute(sql, params)

                if fetch and cursor.description:
                    data = cursor.fetchall()
                    return pd.DataFrame(data)

                return affected_rows

            except pymysql.Error as e:
                raise DBConnectionError(f"Execute failed: {str(e)}") from e